package com.example.mq.mqserver.core;

import com.example.mq.common.mode.Consumer;
import com.example.mq.common.mode.ConsumerEnv;
import com.example.mq.common.exception.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 该类负责实现消息消费的核心逻辑。
 */
@Slf4j
public class ConsumerManager {
    // 引用上层的 VirtualHost 对象，用于操作数据
    private final VirtualHost parent;

    // 线程池，负责执行具体的回调任务
    private final ExecutorService workerPool = Executors.newFixedThreadPool(4);

    // 消息更新后待处理队列
    private final BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    /**
     * 构造函数，初始化 ConsumerManager 并启动一个后台扫描线程
     *
     * @param p VirtualHost 对象的引用
     */
    public ConsumerManager(VirtualHost p) {
        parent = p;

        // 创建并启动扫描线程
        Thread scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 从 tokenQueue 中获取队列名（令牌）
                    String queueName = tokenQueue.take();

                    // 2. 根据队列名获取队列对象
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现, 该队列名不存在! queueName=" + queueName);
                    }

                    // 3. 消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    log.error(e.getMessage());
                }
            }
        });

        // 设置线程为后台线程并启动
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 当有新消息发送时，调用此方法通知消费消息
     *
     * @param queueName 队列名
     * @throws InterruptedException 如果线程被中断
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName); // 将队列名加入 tokenQueue，通知消费者
    }

    /**
     * 为指定的队列添加订阅者
     *
     * @param consumerTag 订阅者标识
     * @param queueName   队列名
     * @param autoAck     是否自动应答
     * @param consumer    消费者对象
     * @throws MqException 如果队列不存在
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 查找队列对象
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }

        // 创建一个订阅者对象并注册到队列中
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);

            // 检查队列中是否有消息，将消息全部发送给订阅者
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumeMessage(queue);
            }
        }
    }

    /**
     * 消费队列中的一条消息
     *
     * @param queue 消息队列
     */
    private void consumeMessage(MSGQueue queue) {
        // 1. 选择一个订阅者，采用轮询的方式
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 如果当前没有订阅者，则不进行消费
            return;
        }

        // 2. 从队列中取出一条消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 如果队列中没有消息，则不进行消费
            return;
        }

        // 3. 将消息传递给订阅者的回调方法，并交给线程池执行
        workerPool.submit(() -> {
            try {
                // 1. 将消息添加到待确认集合中，确保在执行回调之前进行
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);

                // 2. 执行订阅者的回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(),
                        message.getBody());

                // 3. 根据应答模式处理消息删除逻辑
                if (luckyDog.isAutoAck()) {
                    // 自动应答模式，删除消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message); // 删除硬盘上的消息
                    }
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId()); // 删除待确认集合中的该消息的记录
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId()); // 删除内存中的消息
                    log.info("[ConsumerManager] 消息被成功消费! queueName={}", queue.getName());
                }
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        });
    }
}
